#include <stdio.h>
#include "block_queue.h"

queue_t * queue_middle(queue_t *queue){
    queue_t  *middle, *next;

    middle = queue_head(queue);
    if (middle == queue_last(queue)) {
        return middle;
    }

    next = queue_head(queue);
    for ( ;; ) {
        middle = queue_next(middle);

        next = queue_next(next);
        if (next == queue_last(queue)) {
            return middle;
        }

        next = queue_next(next);
        if (next == queue_last(queue)) {
            return middle;
        }
    }
}

int queue_size(queue_t *q){
   if(!q) return -1;
   int size = 0;
   queue_t *p = queue_head(q);
   for( ;p != queue_sentinel(q); p = queue_next(p) ){
       size++;
   }
   return size;
}

/* the stable insertion sort */
void
queue_sort(queue_t *queue,
    int (*cmp)(const queue_t *, const queue_t *))
{
    queue_t  *q, *prev, *next;

    q = queue_head(queue);
    if (q == queue_last(queue)) {
        return;
    }

    for (q = queue_next(q); q != queue_sentinel(queue); q = next) {
        prev = queue_prev(q);
        next = queue_next(q);
        queue_remove(q);

        do {
            if (cmp(prev, q) <= 0) {
                break;
            }
            prev = queue_prev(prev);
        } while (prev != queue_sentinel(queue));

        queue_insert_after(prev, q);
    }
}

void queue_destory(queue_t *q,free_func func){
   if(!q || !q->next || queue_empty(q))
      return;
   if(!func){
      printf("free_func can`t be NULL!\n");
      return;
      //*func = free;
    }
    queue_t *next;
    queue_t *p= queue_head(q);
    for(;p != queue_sentinel(q); p = next){
        next = queue_next(p);
        //queue_remove(p);
        (func)(p);
    }
    //(*func)(q);
}

void block_queue_destory(block_queue_t *bq,free_func func){
    if(!bq ||  !bq->queue.next){
      return ;
    }
    queue_destory(&bq->queue,func);
    queue_init(&bq->queue);
    bq->size = 0;
    pthread_mutex_destroy(&bq->mutex);
    pthread_cond_destroy(&bq->cond);
}

void block_queue_init(block_queue_t *bq,int max_size,unsigned short block_flag){
    if(!bq) return;
    bq->size = 0;
    bq->max_size = max_size;
    bq->block_flag = block_flag;
    queue_init(&bq->queue);
    pthread_mutex_init(&bq->mutex,NULL);
    pthread_cond_init(&bq->cond,NULL);
}

//Notice: can`t add the same entity twice
int block_queue_push(block_queue_t *bq,queue_t *entity){
    printf("%ld thread push bq->size:%d bq->block_flag:%d\n",pthread_self(),bq->size,bq->block_flag);
    pthread_mutex_lock(&bq->mutex);
    if(entity == queue_sentinel(&bq->queue)){
       printf("%ld thread the pg is empty!",pthread_self());
       pthread_mutex_unlock(&bq->mutex);
       return bq->size;
    }
    int arriveMaxCnt = 0;
    while(bq->max_size && bq->size >= bq->max_size){// arrive at max limit
      arriveMaxCnt++;
      printf("bq arrive at max size limit :%d arriveMaxCnt:%d bq->block_flag:%d thread_id:%ld\n",
		bq->max_size,arriveMaxCnt,bq->block_flag,pthread_self());
      //pthread_cond_signal(&bq->cond); //
      if(!bq->block_flag){
         pthread_mutex_unlock(&bq->mutex);
	     return -1;
      }
      struct timespec ts;
      time_t ti;
      time(&ti);
      int sec = (arriveMaxCnt-1)*2 + 1;
      sec = sec > 15 ? 15 : sec;
      ts.tv_sec = ti+sec;
      ts.tv_nsec = 0;//1000*1000*1000*5;
      printf("timewait thread_id:%ld\n", pthread_self());
      pthread_cond_timedwait(&bq->cond,&bq->mutex,&ts);
    }

    queue_insert_head(&bq->queue,entity);
    bq->size++;
    pthread_cond_signal(&bq->cond);
    pthread_mutex_unlock(&bq->mutex);
    return bq->size;
}

queue_t * block_queue_pop(block_queue_t *bq){
    printf("%ld thread pop bq->size:%d bq->block_flag:%d\n",pthread_self(),bq->size,bq->block_flag);
    pthread_mutex_lock(&bq->mutex);
    //while(queue_empty(&bq->queue)){
    while(bq->size <= 0){ // for the same entity twice error
        printf("%ld thread pop bq size:%d flag:%d\n",pthread_self(),bq->size,bq->block_flag);
        if(!bq->block_flag){
           pthread_mutex_unlock(&bq->mutex);
	       return NULL;
        }
        printf("%ld thread pop bq size:%d flag:%d to waiting\n",pthread_self(),bq->size,bq->block_flag);
        pthread_cond_wait(&bq->cond,&bq->mutex);
    }
    queue_t *en = queue_last(&bq->queue);
    queue_remove(en);
    bq->size--;
    pthread_cond_signal(&bq->cond);
    pthread_mutex_unlock(&bq->mutex);
    return en;
}

void broadcast_threads_wakeup(block_queue_t *bq,unsigned short block_flag){
	pthread_mutex_lock(&bq->mutex);
	bq->block_flag = block_flag;
	pthread_cond_broadcast(&bq->cond);
	pthread_mutex_unlock(&bq->mutex);
}
